package com.shujia.youhua

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

/**
 * repartition:coalesce(numPartitions，true) 增多分区使用这个,会产生shuffle
 *
 * coalesce(numPartitions，false) 减少分区 没有shuffle只是合并 partition
 *
 * 当集群资源充足的时候,可以适当的增加分区,增加rdd的并行度,增加任务数,更快执行,repartition
 * 当集群资源紧张的时候,可以适当的减少分区,降低rdd的并行度,减少任务数 coalesce
 * repartition底层就是通过coalesce实现的,通过coalesce来去调整是否产生shuffle
 */
object RepartitionDemo {
  def main(args: Array[String]): Unit = {
    /**
     *  repartition
     *  增大分区，会产生shuffle，前一个RDD分区中的数据会被分散到后一个RDD中不同的分区去
     *  减小分区,也会产生shuffle,会对数据进行适当的排序
     *
     */
    val ss: SparkSession = SparkSession.builder()
      .master("local")
      .appName("缓存，避免使用重复的RDD")
      .getOrCreate()
    val sc: SparkContext = ss.sparkContext

    val lineRDD: RDD[String] = sc.textFile("spark/data/students.txt")
    println(s"===========lineRDD的分区数为：${lineRDD.getNumPartitions}==================")

    //使用repartition，将分区变成10个分区,会得到一个新的RDD
    val rdd1: RDD[String] = lineRDD.repartition(10)
    println(s"===========rdd1的分区数为：${rdd1.getNumPartitions}==================")
//    rdd1.foreach(println)

    //使用repartition，将分区变成1个分区,会得到一个新的RDD
    val rdd2: RDD[String] = lineRDD.repartition(1)
    println(s"===========rdd2的分区数为：${rdd2.getNumPartitions}==================")
    rdd2.foreach(println)

    while (true){

    }


  }
}
